#ifndef ACE_ATCP_CONNECTION_H
#define ACE_ATCP_CONNECTION_H

#include "ace/Asynch_IO.h"
#include "ace/INET_Addr.h" 
#include <ace/Message_Block.h>

#include "tcp_connection.h"
#include "../tcp_session.h"
#include "detail/tcp_session_accessor.h"
#include <cassert>

namespace vpm
{
	namespace mina
	{
		namespace detail
		{

            template <class Session,class SessionConfig>
			class AceATcpConnection  : public ACE_Service_Handler, public TcpConnection
			{
			public:
				enum 
				{
					TCP_MESSAGE_MAX_LENGTH = 4096,
					TCP_DATA_BUFFER_LENGTH = 8192
				};
				AceATcpConnection(Session * pTcpSession=NULL,ConnectionManager* pConnManager= NULL)
                    :TcpConnection(pTcpSession,pConnManager),m_rcvMsgBlock(TCP_DATA_BUFFER_LENGTH)
				{
					TcpSessionAccessor::set_tcp_connection(pTcpSession,NULL); 
					m_sockHandle = ACE_INVALID_HANDLE;
					m_bCloseFlag = true; 
				}

				virtual ~AceATcpConnection()
				{
					if (this->m_sockHandle != ACE_INVALID_HANDLE)
					{
						ACE_OS::closesocket(this->m_sockHandle);
					}		
				}

				// This is called after the new connection has been accepted.
				virtual void open (ACE_HANDLE handle,ACE_Message_Block &msgBlock)
				{
					m_sockHandle = handle; 

					if (this->m_asynchWrite.open (*this) == -1)
					{
						ACE_ERROR ((LM_ERROR,"%p\n","ACE_Asynch_Write_File::open"));
						this->on_disconnected(); 	
						return;
					}

					// Initiate <ACE_Asynch_Read_Stream>.
					if (this->m_asynchRead.open (*this) == -1)
					{
						ACE_ERROR ((LM_ERROR,"%p\n", "ACE_Asynch_Read_Stream::open"));
						this->on_disconnected(); 
						return;
					}


					if (this->m_asynchRead.read (m_rcvMsgBlock, m_rcvMsgBlock.space ()) != 0)
					{		
						this->on_disconnected(); 
						return ; 			
					}

					//ignore the SIGPIPE 
					ACE_Sig_Action no_sigpipe ((ACE_SignalHandler) SIG_IGN);
					ACE_Sig_Action original_action;
					no_sigpipe.register_action (SIGPIPE, &original_action);

					//no_sigpipe.restore_action (SIGPIPE, original_action);
				}

				virtual Session::TCP_DEMARCATE_RESULT demarcate_data(const char * pData, int dataLen, int &pkgLen) 
				{
					pkgLen = dataLen; 
					return Session::TCP_DEMARCATE_OK ; 
				}

				virtual int process_data(const char * pData, int dataLen)
				{
					assert (m_pTcpSession!= NULL);
					return m_pTcpSession->process_data(pData,dataLen); 	
				}

				virtual void on_connected(int port, const std::string & host) 
				{
					assert (m_pTcpSession!= NULL);
					TcpSessionAccessor::on_connect_ready(m_pTcpSession,port,host,this); 		
				}

				virtual void on_disconnected()
				{
					this->destroy_connection();
				}

				int send_data(const char * pData, int dataLen)
				{
					if (!m_bCloseFlag)
					{
						return 0; 
					}

					ACE_Message_Block *pMsgBlock = new ACE_Message_Block( dataLen); 
					pMsgBlock->copy(pData,dataLen);

					//	pMsgBlock->wr_ptr((size_t)dataLen);
					if (this->m_asynchWrite.write (*pMsgBlock, pMsgBlock->length ()) != 0)
					{
						ACE_ERROR ((LM_ERROR,ACE_TEXT ("%p\n"),	ACE_TEXT ("starting write")));
						pMsgBlock->release ();
						this->on_disconnected(); 
						return -1; 
					}	
					return 0; 
				}


				void close()
				{
					ACE_DEBUG((LM_DEBUG,ACE_TEXT("Connection %p\n"),ACE_TEXT("close")));
					if (m_bCloseFlag)
					{
						m_bCloseFlag = false; 
						if (this->m_sockHandle != ACE_INVALID_HANDLE)
						{
							ACE_OS::shutdown(this->m_sockHandle, ACE_SHUTDOWN_WRITE);				
						}				
					}	
				}

				virtual void addresses (const ACE_INET_Addr &remote_address, const ACE_INET_Addr &local_address)
				{
					ACE_DEBUG ((LM_DEBUG,ACE_TEXT ("Connection from %s:%d\n"),
                                remote_address.get_host_addr(), remote_address.get_port_number()));
					this->on_connected(remote_address.get_port_number(),remote_address.get_host_addr());
				}

			protected:
				virtual void close_connection()
				{
					assert (m_pTcpSession!= NULL);
					m_pTcpSession->on_connect_close(); 	
					this->close();
				}

				// These methods are called by the framework
				// This is called when asynchronous <read> operation from the socket
				// complete.
				virtual void handle_read_stream (const ACE_Asynch_Read_Stream::Result &result)
				{
					ACE_Message_Block &msgBlock = result.message_block ();

					if (!result.success () || result.bytes_transferred () == 0)
					{			
#if defined (ACE_WIN32)
						if (result.error () == ERROR_OPERATION_ABORTED)
						{
							return ;
						}
#else
						if (result.error () == ECANCELED)
						{
							return ;
						}
#endif /* ACE_WIN32 */
						this->on_disconnected();			
					}
					else
					{
						if (m_pTcpSession == NULL)
						{
							return ; 
						}

						ACE_DEBUG((LM_DEBUG,ACE_TEXT("Received data length :%d"),msgBlock.length()));

						int pkgSize = 0; 
						while (m_rcvMsgBlock.length() > 0) 
						{
							Session::TCP_DEMARCATE_RESULT  ret = m_pTcpSession->demarcate_data(m_rcvMsgBlock.rd_ptr(),
                                    (int)m_rcvMsgBlock.length(), pkgSize);
							if (ret == Session::TCP_DEMARCATE_OK)
							{
								m_pTcpSession->process_data(m_rcvMsgBlock.rd_ptr(),pkgSize); 
								assert(pkgSize >0);
								m_rcvMsgBlock.rd_ptr(pkgSize);
							}
							else if (ret == Session::TCP_DEMARCATE_INCOMPLETE)
							{				 
								//if (m_rcvMsgBlock.space() == 0)
								//{
								//    m_rcvMsgBlock.crunch();
								//    //m_rcvMsgBlock.reset();				
								//}
								break;
							}
							else 
							{
								//drop the message
								m_rcvMsgBlock.reset();				
								break;
							}
						}
						m_rcvMsgBlock.crunch();

						this->m_asynchRead.read (m_rcvMsgBlock, m_rcvMsgBlock.space ());
					}
				}

				// This is called when an asynchronous <write> to the socket
				// completes.
				virtual void handle_write_stream(const ACE_Asynch_Write_Stream::Result &result)
				{
					ACE_Message_Block &mb = result.message_block ();
					if (!result.success () || result.bytes_transferred () == 0)
					{
						mb.release ();
						this->close();
					}
					else 
					{
						if (mb.length() >0 )
						{
							if (this->m_asynchWrite.write (mb, mb.length ()) != 0)
							{
								ACE_ERROR ((LM_ERROR,ACE_TEXT ("%p\n"),	ACE_TEXT ("starting write left data")));
								mb.release ();
								this->close(); 
							}
						}
						else 
						{
							mb.release ();
						}
					}
				}

			private:
				ACE_HANDLE m_sockHandle;
				bool m_bCloseFlag; 
				ACE_Message_Block  m_rcvMsgBlock; 

				ACE_Asynch_Read_Stream m_asynchRead;

				ACE_Asynch_Write_Stream m_asynchWrite;
			};

#define TcpConnectionImpl AceATcpConnection
		}
	}
}

#endif //  __ACE_ATCP_CONNECTION_HPP__

